package com.yc.streaming.util;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yc.streaming.common.Constants;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import scala.Tuple2;


public final class RedisUtils {
	private static Logger logger = LoggerFactory.getLogger(RedisUtils.class);

	// Redis服务器IP
	public static String ADDR = ConfigurationManager.getProperty(Constants.REDIS_HOST);

	// Redis的端口号
	public static int PORT = ConfigurationManager.getInteger(Constants.REDIS_PORT);


	// 可用连接实例的最大数目，默认值为8；
	// 如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis实例，则此时pool的状态为exhausted(耗尽)。
	public static int MAX_ACTIVE = 1000;

	// 控制一个pool最多有多少个状态为idle(空闲的)的jedis实例，默认值也是8。
	public static int MAX_IDLE = 8;

	// 等待可用连接的最大时间，单位毫秒，默认值为-1，表示永不超时。如果超过等待时间，则直接抛出JedisConnectionException；
	public static int MAX_WAIT = 100000;

	public static int TIMEOUT = 50000;

	// 在borrow一个jedis实例时，是否提前进行validate操作；如果为true，则得到的jedis实例均是可用的；
	public static boolean TEST_ON_BORROW = true;

	private static JedisPool jedisPool = null;

	/**
	 * Jedis操作步骤如下： 
	 * 1->获取Jedis实例需要从JedisPool中获取； 
	 * 2->用完Jedis实例需要返还给JedisPool；
	 * 3->如果Jedis在使用过程中出错，则也需要还给JedisPool；
	 */

	/**
	 * 初始化Redis连接池
	 */
	static {
		try {
			JedisPoolConfig config = new JedisPoolConfig();
			config.setMaxTotal(MAX_ACTIVE);
			config.setMaxIdle(MAX_IDLE);
			config.setMaxWaitMillis(MAX_WAIT);
			config.setTestOnBorrow(TEST_ON_BORROW);
			jedisPool = new JedisPool(config, ADDR, PORT, TIMEOUT);
		} catch (Exception e) {
			logger.error("init jedispool error", e);
		}
	}

	public static JedisPool getPool() {
		return jedisPool;
	}

	/**
	 * 获取Jedis实例
	 * 
	 * @return
	 */
	public synchronized static Jedis getJedis() {
		try {
			if (jedisPool != null) {
				Jedis resource = jedisPool.getResource();
				return resource;
			} else {
				return null;
			}
		} catch (Exception e) {
			logger.error("getJedis error", e);
			return null;
		}
	}

	/**
	 * 释放jedis资源
	 * 
	 * @param jedis
	 */
	public static void returnResource(final Jedis jedis) {
		if (jedis != null) {
			jedis.close();
		}
	}

	/**
	 * 
	 * @param map
	 * @param indexDb
	 * @param key
	 */
	public static void setAll(Map<String, String> map, int indexDb, String key) {
		Jedis redis = null;
		try {
			redis = RedisUtils.getJedis();
			redis.select(indexDb);
			redis.hmset(key, map);
		} catch (Exception e) {
			logger.error("setAll error", e);
			throw e;
		} finally {
			if (redis != null) {
				RedisUtils.returnResource(redis);
			}
		}
	}

	/**
	 *  添加带生存时间的key
	 */
	public static void setAllWithEx(Map<String, String> map, int indexDb,int time, String prefix) {
		Jedis redis = null;
		try {
			redis = RedisUtils.getJedis();
			redis.select(indexDb);
			for (String key : map.keySet()) {
				redis.setex(prefix + "_" + key, time, key + "::" + map.get(key));
			}
		} catch (Exception e) {
			logger.error("setAllWith Error ", e);
		} finally {
			if (redis != null) {
				RedisUtils.returnResource(redis);
			}
		}
	}

	/**
	 *  添加带生存时间的key
	 */
	public static void setAllWithEx( int indexDb,Set<String> keySet,int time, String prefix) {
		Jedis redis = null;
		try {
			if(keySet == null || keySet.isEmpty())
				return;
			redis = RedisUtils.getJedis();
			redis.select(indexDb);
			for (String key : keySet) {
				redis.setex(prefix + key, time, "1");
			}
		} catch (Exception e) {
			logger.error("setAllWith Error ", e);
		} finally {
			if (redis != null) {
				RedisUtils.returnResource(redis);
			}
		}
	}
	
	/**
	 * 根据pattern获取所有的value
	 * 
	 * @param indexDb
	 * @param keyPattern
	 * @return
	 */
	public static Map<String, String> getAllWithEx(int indexDb,String keyPattern) {
		Jedis redis = null;
		Map<String, String> result = new HashMap<String, String>();
		try {
			redis = RedisUtils.getJedis();
			redis.select(indexDb);
			Set<String> values = redis.keys(keyPattern);
			Iterator<String> kvs = values.iterator();
			while (kvs.hasNext()) {
				String tmp = kvs.next();
				if(tmp!=null&&redis.get(tmp)!=null){
					String[] kv = redis.get(tmp).split("::");
					String key = kv[0];
					String value = kv[1];
					result.put(key, value);
				}else{
					result.put("", "");
				}
			}
			return result;
		} catch (Exception e) {
			logger.error("getAllWithEX error", e);
			throw e;
		} finally {
			if (redis != null) {
				RedisUtils.returnResource(redis);
			}
		}
	}

	/**
	 * 根据pattern获取所有的value
	 * 
	 * @param indexDb
	 * @param keyPattern
	 * @return
	 */
	public static Set<String> getAllKeys(int indexDb,String keyPattern) {
		Jedis redis = null;
		try {
			redis = RedisUtils.getJedis();
			redis.select(indexDb);
			return redis.keys(keyPattern);
		} catch (Exception e) {
			logger.error("getAllWithEX error", e);
			return Collections.emptySet();
		} finally {
			if (redis != null) {
				RedisUtils.returnResource(redis);
			}
			
		}
	}
	
	/**
	 * 
	 * @param indexDb
	 * @param key
	 * @return
	 */
	public static Map<String, String> getAll(int indexDb, String key,boolean isFlush) {
		Jedis redis = null;
		try {
			redis = RedisUtils.getJedis();
			redis.select(indexDb);
			Map<String,String> map = redis.hgetAll(key);
			if(isFlush){
				redis.flushDB();
			}
			return map;
		} catch (Exception e) {
			logger.error("getAll error", e);
			throw e;
		} finally {
			if (redis != null) {
				RedisUtils.returnResource(redis);
			}
		}
	}
	

	public static Map<String, String> getAll(int indexDb, String key,List<String> fields) {
		Jedis redis = null;
		try {
			Map<String, String> result = new HashMap<String, String>(fields.size());
			redis = RedisUtils.getJedis();
			redis.select(indexDb);
			for (int i = 0; i < fields.size(); i++) {
				result.put(fields.get(i), redis.hmget(key, fields.get(i)).toString().trim());
			}
			return result;
		} catch (Exception e) {
			logger.error("getAll error", e);
			throw e;
		} finally {
			if (redis != null) {
				RedisUtils.returnResource(redis);
			}
		}
	}
	
	// 将Map转换为 元素是tuple2的list
	public static List<Tuple2<String, String>> getList(Map<String, String> map) {
		List<Tuple2<String, String>> list = new ArrayList<Tuple2<String, String>>();
		for (String key : map.keySet()) {
			if(!"".equals(map.get(key))){
				list.add(new Tuple2<String, String>(key, map.get(key)));
			}
		}
		return list;
	}
}